package lineage

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.junit.{After, Before, Test}

/**
 * 除了窄依赖的依赖都是宽依赖
 */
class Driver {

	val conf: SparkConf = new SparkConf().setAppName("Serializ").setMaster("local[3]")
	var sc: SparkContext = null
	var outpath: String = "out"
	import util.MyPredef._

	@Before
	def init() {
		sc = new SparkContext(conf)
		outpath.delete()
	}

	@After
	def after() {
		sc.stop()
	}

	@Test
	def main(): Unit = {
		val word = sc.textFile("in").flatMap(_.split(" ")).map((_, 1))
		val value: RDD[(String, Int)] = word.reduceByKey(_ + _)
		value.collect()
		println(word.toDebugString) // 行动算子执行才会触发checkpoint
		println("----------------")
		println(value.toDebugString) // 行动算子执行才会触发checkpoint
	}
}
